package demo.kafka.streams.cfg;

import com.jucoin.infra.utils.LaneEnvUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

import java.util.Map;
import java.util.Set;

/**
 * 泳道管理配置类
 */
//@Configuration
public class StreamsLaneConfig {

    private static final Logger logger = LoggerFactory.getLogger(StreamsLaneConfig.class);

    /**
     *
     *
     * @param original
     * @return
     */
    @Bean
    @Primary
    public BindingServiceProperties bindingServiceProperties(BindingServiceProperties original) {
        String laneTag = LaneEnvUtil.getLaneTag();
        Set<String> laneTopics = LaneEnvUtil.getLaneTopics();
        logger.info("泳道laneTag: {} , 隔离laneTopics: {}", laneTag, laneTopics);
        if (laneTag == null || laneTag.isEmpty()) {
            return original;
        }
        if (laneTopics == null || laneTopics.isEmpty()) {
            return original;
        }
        Map<String, BindingProperties> bindings = original.getBindings();
        for (Map.Entry<String, BindingProperties> entry : bindings.entrySet()) {
            String bindingName = entry.getKey();
            BindingProperties bindingProperties = entry.getValue();
            String topicName = bindingProperties.getDestination();
            if (topicName == null || topicName.isEmpty()) {
                logger.warn("Binding {} 的destination为空，跳过泳道处理", bindingName);
                continue;
            }
            String laneTopic = LaneEnvUtil.getLaneKafkaTopic(topicName);
            logger.info("bindingName: {} , 当前topicName: {} , 对应泳道laneTopic: {}", bindingName, topicName, laneTopic);
            bindingProperties.setDestination(laneTopic);
        }
        return original;
    }


    /**
     * 配置Kafka Streams应用的ApplicationId，添加泳道环境标识
     *
     * @param original
     * @return
     */
    @Bean
    @Primary  // 确保这个Bean优先被使用
    public KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties(
            KafkaStreamsBinderConfigurationProperties original) {
        String laneTag = LaneEnvUtil.getLaneTag();
        Set<String> laneTopics = LaneEnvUtil.getLaneTopics();
        logger.info("laneTag: {} , laneTopics: {}", laneTag, laneTopics);
        if (laneTag == null || laneTag.isEmpty()) {
            return original;
        }
        if (laneTopics == null || laneTopics.isEmpty()) {
            return original;
        }
        String appId = original.getApplicationId();
        if (appId == null || appId.isEmpty()) {
            return original;
        }
        String newAppId = appId + "_" + laneTag;
        original.setApplicationId(newAppId);
        logger.info("laneTag: {} , newAppId: {}", laneTag, newAppId);
        return original;
    }
}